package product;


import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Shanxi {

/**计算山西省的每种农产品的价格波动趋势， 即计算每天价格均值， 并按照时间先后顺序排列该值。
某种农产品的价格均值计算公式：PAVG = (PM1+PM2+...+PMn-max(P)-min(P))/(N-2)
其中， P 表示价格， Mn 表示 market， 即农产品市场。 PM1 表示 M1 农产品市场的该产品价格， max(P)表示价格最大值， min(P)价格最小值。
* @param args
*/
	@SuppressWarnings("deprecation")
	public static void main(String[] args) throws Exception{
		
		if (args.length!=2) {
			System.err.println("path ERR!");
			System.exit(-1);
		}
		Job job=new Job(new Configuration(), "PAVG");
		job.setJarByClass(Shanxi.class);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		job.setMapperClass(avmaps.class);
		job.setReducerClass(avreduce.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(DoubleWritable.class);
		job.waitForCompletion(true);
	}
	public static class avmaps extends Mapper<LongWritable, Text,Text, DoubleWritable>{
		protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,Text,DoubleWritable>.Context context) throws java.io.IOException ,InterruptedException {
			String[] lines = value.toString().split("\t");
			if (lines.length==6 && lines[4].equals("山西")) {
				double price=Double.parseDouble(lines[1].trim());
				context.write(new Text(lines[0].trim()), new DoubleWritable(price));
			}
		};
	}
	
	public static class avreduce extends Reducer<Text, DoubleWritable, Text, DoubleWritable>{
		int N=0;
		double pavg=0;
		double maxtype=0;
		double mintype=0;
		List<Double> list=new ArrayList<>();
		protected void reduce(Text k2, java.lang.Iterable<DoubleWritable> values, org.apache.hadoop.mapreduce.Reducer<Text,DoubleWritable,Text,DoubleWritable>.Context context, double maxtype) throws java.io.IOException ,InterruptedException {
			double sum=0;
			for (DoubleWritable price : values) {
				sum+=price.get();
				N++;
				list.add(price.get());
			}
			maxtype=Collections.max(list);
			mintype=Collections.min(list);
			if (N>2) {
				pavg=(sum-maxtype-mintype)/(N-2);
				context.write(k2, new DoubleWritable(pavg));
			}else if(N>0){
				pavg=sum/N;
				context.write(k2, new DoubleWritable(pavg));
			}
		};
	}
	
}
